//********************************************************************
//	created:	15:10:2011   4:55
//	filename: 	apnservice.cpp
//	author:		tiamo
//	purpose:	apn service
//********************************************************************

#include "stdafx.h"
#include "apnservice.h"

namespace apn
{
	namespace sql
	{
		//
		// load notification list
		//
		class LoadNotificationList : public mysql::SQLRequest
		{
		public:
			//
			// constructor
			//
			LoadNotificationList(mysql::Database& db) : SQLRequest(db, L"LoadNotificationList") {}

		private:
			//
			// do request imp
			//
			virtual void doRequestImp();

		public:
			//
			// notification list
			//
			std::vector<notification>										mNotificationList;

			//
			// dev mode
			//
			bool															mDevMode;
		};

		//
		// save push result
		//
		class SaveResult : public mysql::SQLRequest
		{
		public:
			//
			// constructor
			//
			SaveResult(mysql::Database& db) : SQLRequest(db, L"SaveRequest") {}

		private:
			//
			// do request imp
			//
			virtual void doRequestImp();

		public:
			//
			// id
			//
			uint32_t														mId;

			//
			// status
			//
			uint8_t															mStatus;
		};
		
		//
		// deactive device
		//
		class DeactiveDevice : public mysql::SQLRequest
		{
		public:
			//
			// constructor
			//
			DeactiveDevice(mysql::Database& db) : SQLRequest(db, L"DeactiveDevice"){}

		private:
			//
			// do request imp
			//
			virtual void doRequestImp();

		public:
			//
			// device token
			//
			UniString														mDeviceToken;

			//
			// delete time
			//
			uint32_t														mDeleteTime;
		};
	}

	//
	// initialize
	//
	void Service::initialize()
	{
		//
		// call super
		//
		::Service::initialize();

		//
		// load config
		//
		utils::Config* cfg													= Server::getSingleton()->getConfig();
		UniString configPath												= UniString(L"/config/apn/") + (mDevMode ? L"development/" : L"production/");
		mCARoot																= cfg->getString(configPath + L"ca_root", L"");
		mApnHost															= cfg->getString(configPath + L"apn_host", mDevMode ? L"gateway.sandbox.push.apple.com" : L"gateway.push.apple.com");
		mApnPort															= cfg->getInt(configPath + L"apn_port", 2195);
		mFeedbackHost														= cfg->getString(configPath + L"feedback_host", mDevMode ? L"feedback.sandbox.push.apple.com" : L"feedback.push.apple.com");
		mFeedbackPort														= cfg->getInt(configPath + L"feedback_port", 2196);
		mClientCert															= cfg->getString(configPath + L"cert");
		mClientKey															= cfg->getString(configPath + L"key", mClientCert.c_str());

		//
		// create anp/feedback socket
		//
		createSocket(true, 10);
		createSocket(false, 10);

		//
		// setup timer
		//
		uint32_t delayTime													= 60;
		LogInfo(L"anp::Service(%p:%ls): setup push timer(%u)", this, mName.c_str(), delayTime);
		Server::getSingleton()->getTimerManager()->registerTimer(delayTime, false, std::bind(&Service::onPushNotificationTimer, this, std::placeholders::_1));
		LogInfo(L"apn::Service(%p:%ls): running (%ls)=(%ls) in (%ls) mode", this, mName.c_str(), mName.c_str(), mClientCert.c_str(), mDevMode ? L"development" : L"production");
	}

	//
	// create socket
	//
	void Service::createSocket(bool apnSocket, uint32_t delayTime)
	{
		//
		// create immediately
		//
		if(!delayTime)
		{
			UniString const& remoteHost										= apnSocket ? mApnHost : mFeedbackHost;
			uint32_t remotePort												= apnSocket ? mApnPort : mFeedbackPort;
			LogInfo(L"apn::Service(%p:%ls): creating %ls socket to (%ls:%u)", this, mName.c_str(), apnSocket ? L"apn" :L"feedback", remoteHost.c_str(), remotePort);

			try
			{
				utils::SmartPointer<network::SSLSocket> theSocket			= new network::SSLSocket;
				theSocket->setSSLInfo(mCARoot, mClientKey, mClientKey);
				theSocket->setUserData(reinterpret_cast<void*>(apnSocket));
				theSocket->setConnectCallback(std::bind(&Service::onSocketConnected, this, std::placeholders::_1, std::placeholders::_2));
				theSocket->setCloseCallback(std::bind(&Service::onSocketClosed, this, std::placeholders::_1, std::placeholders::_2));
				theSocket->setDecodeDataCallback(std::bind(&Service::onSocketDecodePacket, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4));
				theSocket->setDataPacketCallback(std::bind(&Service::onSocketPacket, this, std::placeholders::_1, std::placeholders::_2));
				if(theSocket->createTcpClientSocket(remoteHost, static_cast<uint16_t>(remotePort), true) && Server::getSingleton()->getSocketManager()->addSocket(theSocket.get()))
				{
					if(apnSocket)
						mApnSocket											= theSocket.get();
					else
						mFeedbackSocket										= theSocket.get();

					return;
				}
			}
			catch(std::exception& e)
			{
				LogError(L"apn::Service(%p:%ls): creating %ls socket to (%ls:%u) failed(%ls)", this, mName.c_str(), apnSocket ? L"apn" :L"feedback", remoteHost.c_str(), remotePort, utils::Exception::getMessage(e).c_str());
			}
			
			LogError(L"apn::Service(%p:%ls): creating %ls socket to (%ls:%u) failed, scheduling reconnect timer", this, mName.c_str(), apnSocket ? L"apn" :L"feedback", remoteHost.c_str(), remotePort);
			delayTime														= 5;
		}

		auto delayTimer = [this, apnSocket](uint32_t elapsedTime)
		{
			createSocket(apnSocket);
			return true;
		};
		LogInfo(L"apn::Service(%p:%ls): delay create %ls socket", this, mName.c_str(), apnSocket ? L"apn" :L"feedback");
		Server::getSingleton()->getTimerManager()->registerTimer(delayTime, true, std::move(delayTimer));
	}

	//
	// push notification timer
	//
	bool Service::onPushNotificationTimer(uint32_t elapsedTime)
	{
		LogInfo(L"apn::Service(%p:%ls): loading notification", this, mName.c_str());
		utils::SmartPointer<sql::LoadNotificationList> loadNotificationList	= new sql::LoadNotificationList(mDatabase);
		loadNotificationList->mDevMode										= mDevMode;
		loadNotificationList->setCallback(std::bind(&Service::onLoadNotification, this, std::placeholders::_1));
		Server::getSingleton()->getSQLRequestManager()->pushPendingRequest(loadNotificationList.get());
		return false;
	}

	//
	// load notification
	//
	void Service::onLoadNotification(utils::BaseRequest* theRequest)
	{
		//
		// check result
		//
		sql::LoadNotificationList* loadNotificationList						= static_cast<sql::LoadNotificationList*>(theRequest);
		LogInfo(L"apn::Service(%p:%ls): load notification(%u,%u) finished(%d)", this, mName.c_str(), static_cast<uint32_t>(loadNotificationList->mNotificationList.size()), static_cast<uint32_t>(mNotificationList.size()), theRequest->getResult());

		//
		// add to list
		//
		std::copy(loadNotificationList->mNotificationList.begin(), loadNotificationList->mNotificationList.end(), std::back_inserter(mNotificationList));

		//
		// check count
		//
		if(mNotificationList.empty() || !mApnSocket)
			return;

		//
		// send
		//
		LogInfo(L"apn::Service(%p:%ls): send notification(%u)", this, mName.c_str(), static_cast<uint32_t>(mNotificationList.size()));
		utils::DataBuffer sendBuffer(16 * 1024);
		for(auto it = mNotificationList.begin(); it != mNotificationList.end(); ++ it)
		{
			notification& item												= *it;
			NotificationPacket packet;
			packet.mDeviceToken												= std::move(item.mDeviceToken);
			packet.mExpiry													= item.mExpire;
			packet.mId														= item.mId;
			packet.mPayload													= std::move(item.mPayload);
			packet.serialize(sendBuffer, false);
			if(sendBuffer.getLength() >= 15 * 1024)
			{
				mApnSocket->sendBuffer(sendBuffer.getBuffer(), sendBuffer.getLength());
				sendBuffer.resize(0);
			}
		}

		if(sendBuffer.getLength())
			mApnSocket->sendBuffer(sendBuffer.getBuffer(), sendBuffer.getLength());
		
		//
		// clear cache
		//
		mNotificationList.clear();
	}

	//
	// save result
	//
	void Service::onSaveResult(utils::BaseRequest* theRequest)
	{
		//
		// check result, ignore error
		//
		sql::SaveResult* saveResult											= static_cast<sql::SaveResult*>(theRequest);
		LogInfo(L"apn::Service(%p:%ls): save result(%u=%u) finished(%d)", this, mName.c_str(), saveResult->mId, saveResult->mStatus, theRequest->getResult());
	}

	//
	// deactive device token
	//
	void Service::onDeactiveDeviceToken(utils::BaseRequest* theRequest)
	{
		//
		// check result, ignore error
		//
		sql::DeactiveDevice* deactiveDevice									= static_cast<sql::DeactiveDevice*>(theRequest);
		LogInfo(L"apn::Service(%p:%ls): deactive device(%ls) finished(%d)", this, mName.c_str(), deactiveDevice->mDeviceToken.c_str(), theRequest->getResult());
	}

	//
	// socket connected
	//
	void Service::onSocketConnected(network::AsyncSocket* theSocket, bool errorState)
	{
		if(errorState)
			LogError(L"apn::Service(%p:%ls): %ls socket connect faield", this, mName.c_str(), theSocket->getUserData() ? L"apn" : L"feedback");
		else
			LogInfo(L"apn::Service(%p:%ls): %ls socket connect succeeded", this, mName.c_str(), theSocket->getUserData() ? L"apn" : L"feedback");
	}

	//
	// socket closed
	//
	void Service::onSocketClosed(network::AsyncSocket* theSocket, bool errorState)
	{
		if(errorState)
			LogError(L"apn::Service(%p:%ls): %ls socket closed with faield status", this, mName.c_str(), theSocket->getUserData() ? L"apn" : L"feedback");
		else
			LogInfo(L"apn::Service(%p:%ls): %ls socket closed with succeeded status", this, mName.c_str(), theSocket->getUserData() ? L"apn" : L"feedback");

		//
		// reconnect
		//
		if(theSocket == mApnSocket)
		{
			mApnSocket														= nullptr;
			createSocket(true, 10);
		}
		else if(theSocket == mFeedbackSocket)
		{
			mFeedbackSocket													= nullptr;
			createSocket(false, 120);
		}
	}

	//
	// socket decode packet
	//
	utils::SmartPointer<network::BasePacket> Service::onSocketDecodePacket(network::AsyncSocket* theSocket, void const* dataBuffer, uint16_t dataLength, uint32_t& usedLength)
	{
		if(theSocket->getUserData())
		{
			usedLength														= ResultResponsePacket::getMessageLength(dataBuffer, dataLength);
			if(!usedLength)
				return nullptr;

			utils::DataBuffer tempBuffer(dataBuffer, usedLength, true);
			utils::SmartPointer<ResultResponsePacket> retValue				= new ResultResponsePacket;
			retValue->serialize(tempBuffer, true);
			return retValue.get();
		}
		
		usedLength															= FeedbackPacket::getMessageLength(dataBuffer, dataLength);
		if(!usedLength)
			return nullptr;

		utils::DataBuffer tempBuffer(dataBuffer, usedLength, true);
		utils::SmartPointer<FeedbackPacket> retValue						= new FeedbackPacket;
		retValue->serialize(tempBuffer, true);
		return retValue.get();
	}

	//
	// socket data
	//
	void Service::onSocketPacket(network::AsyncSocket* theSocket, network::BasePacket* thePacket)
	{
		LogInfo(L"apn::Service(%p:%ls): %ls socket recieved packet(%ls)", this, mName.c_str(), theSocket->getUserData() ? L"apn" : L"feedback", thePacket->getName());
		if(thePacket->getName() == L"apn::ResultResponsePacket")
		{
			ResultResponsePacket const* resultPacket						= static_cast<ResultResponsePacket const*>(thePacket);
			LogInfo(L"apn::Service(%p:%ls): send notification(%u) result(%u)", this, mName.c_str(), resultPacket->mId, resultPacket->mStatus);
			utils::SmartPointer<sql::SaveResult> saveResult					= new sql::SaveResult(mDatabase);
			saveResult->mId													= resultPacket->mId;
			saveResult->mStatus												= resultPacket->mStatus;
			saveResult->setCallback(std::bind(&Service::onSaveResult, this, std::placeholders::_1));
			Server::getSingleton()->getSQLRequestManager()->pushPendingRequest(saveResult.get());
		}
		else if(thePacket->getName() == L"apn::FeedbackPacket")
		{
			FeedbackPacket const* feedbackPacket							= static_cast<FeedbackPacket const*>(thePacket);
			LogWarning(L"apn::Service(%p:%ls): device(%ls) uninstall at (%ls)", this, mName.c_str(), feedbackPacket->mDeviceToken.c_str(), platform::timeToString(feedbackPacket->mTime).c_str());
			utils::SmartPointer<sql::DeactiveDevice> deactiveDevice			= new sql::DeactiveDevice(mDatabase);
			deactiveDevice->mDeleteTime										= feedbackPacket->mTime;
			deactiveDevice->mDeviceToken									= feedbackPacket->mDeviceToken;
			deactiveDevice->setCallback(std::bind(&Service::onDeactiveDeviceToken, this, std::placeholders::_1));
			Server::getSingleton()->getSQLRequestManager()->pushPendingRequest(deactiveDevice.get());
		}
	}

	//
	// load notification list
	//
	void sql::LoadNotificationList::doRequestImp()
	{
		//
		// transaction
		//
		mDatabase.startTransaction();
		auto rollback														= utils::makeResourceHelper([this](){mDatabase.rollbackTransaction();});

		//
		// load list
		//
		notification item;
		mNotificationList.clear();
		mysql::Statement statement											= mDatabase.createStatement();
		statement.prepareV(L"select id, device_token, payload, unix_timestamp(expire) from notification where sent=0 and dev_mode=%d", mDevMode ? 1 : 0);
		statement.bind(L"id", false, item.mId);
		statement.bind(L"device_token", false, item.mDeviceToken, 64);
		statement.bind(L"payload", false, item.mPayload);
		statement.bind(L"expire", false, item.mExpire);
		statement.execute();
		while(statement.fetch())
			mNotificationList.push_back(item);

		//
		// update sent flags
		//
		statement.prepareV(L"update notification set sent=1 where sent=0 and dev_mode=%d", mDevMode ? 1 : 0);
		statement.execute();

		mDatabase.commitTransaction();
		rollback.dismiss();
	}

	//
	// save result
	//
	void sql::SaveResult::doRequestImp()
	{
		mDatabase.startTransaction();
		auto rollback														= utils::makeResourceHelper([this](){mDatabase.rollbackTransaction();});
		mysql::Statement statement											= mDatabase.createStatement();
		statement.prepare(L"update notification set result=? where id=?");
		statement.bind(L"result", true, mStatus);
		statement.bind(L"id", true, mId);
		statement.execute();
		mDatabase.commitTransaction();
		rollback.dismiss();
	}

	//
	// delete device token
	//
	void sql::DeactiveDevice::doRequestImp()
	{
		mDatabase.startTransaction();
		auto rollback														= utils::makeResourceHelper([this](){mDatabase.rollbackTransaction();});
		mysql::Statement statement											= mDatabase.createStatement();
		statement.prepare(L"update subscription set delete_time = from_unixtime(?) where device_token=? and unix_timestamp(active_time) < ?");
		statement.bind(L"delete_time", true, mDeleteTime);
		statement.bind(L"device_token", true, mDeviceToken);
		statement.bind(L"delete_time_condition", true, mDeleteTime);
		statement.execute();
		mDatabase.commitTransaction();
		rollback.dismiss();
	}

	//
	// get notification packet length
	//
	uint32_t NotificationPacket::getMessageLength(void const* dataBuffer, uint32_t dataLength)
	{
		//
		// command(1) + id(4) + expiry(4) + token_length(2) + token(*) + payload_length(2) + payload(*)
		//
		uint32_t const headerLength											= sizeof(uint8_t) + sizeof(uint32_t) + sizeof(uint32_t) + sizeof(uint16_t);
		if(dataLength < headerLength)
			return 0;

		uint16_t tokenLength												= 0;
		memcpy(&tokenLength, static_cast<uint8_t const*>(dataBuffer) + headerLength - sizeof(uint16_t), sizeof(tokenLength));
		tokenLength															= ntohs(tokenLength);
		if(dataLength < headerLength + tokenLength + sizeof(uint16_t))
			return 0;

		uint16_t payloadLength												= 0;
		memcpy(&payloadLength, static_cast<uint8_t const*>(dataBuffer) + headerLength + payloadLength, sizeof(payloadLength));
		payloadLength														= ntohs(payloadLength);
		if(dataLength < headerLength + tokenLength + sizeof(uint16_t) + payloadLength)
			return 0;

		return headerLength + tokenLength + sizeof(uint16_t) + payloadLength;
	}

	//
	// serialize
	//
	void NotificationPacket::serialize(utils::DataBuffer& dataBuffer, bool readFromBuffer)
	{
		utils::DataBuffer::ReadWriteHelper readWriteHelper(dataBuffer, dataBuffer.getLength());
		if(readFromBuffer)
		{
			uint8_t command													= readWriteHelper.read<uint8_t>();
			if(command != 1)
				ThrowException(L"ErrorResponsePacket(%p): invalid command(%u)", this, command);

			mId																= ntohl(readWriteHelper.read<uint32_t>());
			mExpiry															= ntohl(readWriteHelper.read<uint32_t>());
			uint16_t tokenLength											= ntohs(readWriteHelper.read<uint16_t>());
			mDeviceToken													= utils::binaryToHex(static_cast<uint8_t const*>(readWriteHelper.getBuffer()), tokenLength);
			readWriteHelper.seek(tokenLength);
			uint16_t payloadLength											= ntohs(readWriteHelper.read<uint16_t>());
			mPayload														= utils::utf8ToWide(static_cast<char const*>(readWriteHelper.getBuffer()), payloadLength);
		}
		else
		{
			uint16_t tokenLength											= static_cast<uint16_t>(mDeviceToken.length() / 2);
			Utf8String payload												= utils::wideToUtf8(mPayload);
			uint16_t payloadLength											= static_cast<uint16_t>(payload.length());
			readWriteHelper.write<uint8_t>(1);
			readWriteHelper.write<uint32_t>(htonl(mId));
			readWriteHelper.write<uint32_t>(htonl(mExpiry));
			readWriteHelper.write<uint16_t>(htons(tokenLength));
			readWriteHelper.makeSpace(tokenLength);
			utils::hexToBinary(mDeviceToken, static_cast<uint8_t*>(readWriteHelper.getBuffer()), tokenLength);
			readWriteHelper.seek(tokenLength);
			readWriteHelper.write<uint16_t>(htons(payloadLength));
			readWriteHelper.writeBuffer(payload.c_str(), payloadLength);
		}
	}

	//
	// get error response packet length
	//
	uint32_t ResultResponsePacket::getMessageLength(void const* dataBuffer, uint32_t dataLength)
	{
		//
		// command(1) + status(1) + id(4)
		//
		if(dataLength < sizeof(uint8_t) + sizeof(uint8_t) + sizeof(uint32_t))
			return 0;

		return sizeof(uint8_t) + sizeof(uint8_t) + sizeof(uint32_t);
	}

	//
	// serialize
	//
	void ResultResponsePacket::serialize(utils::DataBuffer& dataBuffer, bool readFromBuffer)
	{
		utils::DataBuffer::ReadWriteHelper readWriteHelper(dataBuffer);
		if(readFromBuffer)
		{
			uint8_t command													= readWriteHelper.read<uint8_t>();
			if(command != 8)
				ThrowException(L"ErrorResponsePacket(%p): invalid command(%u)", this, command);

			mStatus															= readWriteHelper.read<uint8_t>();
			mId																= ntohl(readWriteHelper.read<uint32_t>());
		}
		else
		{
			readWriteHelper.write<uint8_t>(8);
			readWriteHelper.write<uint8_t>(mStatus);
			readWriteHelper.write<uint32_t>(htonl(mId));
		}
	}

	//
	// get feedback packet length
	//
	uint32_t FeedbackPacket::getMessageLength(void const* dataBuffer, uint32_t dataLength)
	{
		//
		// time(4) + token_length(2) + token(*)
		//
		if(dataLength < sizeof(uint32_t) + sizeof(uint16_t))
			return 0;

		uint16_t tokenLength												= 0;
		memcpy(&tokenLength, static_cast<uint8_t const*>(dataBuffer) + sizeof(uint32_t), sizeof(tokenLength));
		tokenLength															= ntohs(tokenLength);

		return sizeof(uint32_t) + sizeof(uint16_t) + tokenLength;
	}

	//
	// serialize
	//
	void FeedbackPacket::serialize(utils::DataBuffer& dataBuffer, bool readFromBuffer)
	{
		utils::DataBuffer::ReadWriteHelper readWriteHelper(dataBuffer);
		if(readFromBuffer)
		{
			mTime															= ntohl(readWriteHelper.read<uint32_t>());
			uint16_t tokenLength											= ntohs(readWriteHelper.read<uint16_t>());
			mDeviceToken													= utils::binaryToHex(static_cast<uint8_t const*>(readWriteHelper.getBuffer()), tokenLength);
		}
		else
		{
			uint32_t deviceTokenLength										= static_cast<uint32_t>(mDeviceToken.length() / 2);
			readWriteHelper.write<uint32_t>(htonl(mTime));
			readWriteHelper.write<uint16_t>(htons(deviceTokenLength));
			readWriteHelper.makeSpace(deviceTokenLength);
			utils::hexToBinary(mDeviceToken, static_cast<uint8_t*>(readWriteHelper.getBuffer()), deviceTokenLength);
		}
	}
}
